package com.bw.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.bean.BoundSort;
import com.bw.fun.Mn6AsyncIOFunc;
import com.bw.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Properties;
import java.util.concurrent.TimeUnit;


// todo 水印傳遞有問題
public class Test5_1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<String> stream = env.addSource(MyKafkaUtil.getKafkaConsumer("tms_dwd_bound_sort_yk6", "test676"));


        SingleOutputStreamOperator<String> etlStream = stream.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                try {
                    JSONObject js = JSON.parseObject(s);
                    String sort_time = js.getString("sort_time");
                    if (sort_time != null) {
                        return true;
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
                return false;
            }
        });

//        etlStream.print();

        // 转成2元组
        SingleOutputStreamOperator<Tuple2<String, Long>> timeDS = etlStream.map(new MapFunction<String, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(String s) throws Exception {
                JSONObject js = JSON.parseObject(s);
                return new Tuple2<>(js.getString("org_id"), js.getLong("sort_time"));
            }
            // 把分拣时间当做水位线
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(0)).withTimestampAssigner((event, timestamp) -> event.f1));


        timeDS.print();
        // 开窗聚合

        SingleOutputStreamOperator<BoundSort> process = timeDS.keyBy(x -> x.f0).window(TumblingEventTimeWindows.of(Time.seconds(10)))//,Time.hours(-8)
                .process(new ProcessWindowFunction<Tuple2<String, Long>, BoundSort, String, TimeWindow>() {
                    @Override
                    public void process(String key, ProcessWindowFunction<Tuple2<String, Long>, BoundSort, String, TimeWindow>.Context context, Iterable<Tuple2<String, Long>> iterable, Collector<BoundSort> collector) throws Exception {
                        System.out.println("\"process\" = " + "process");
                        int cnt = 0;
                        for (Tuple2<String, Long> boundSort : iterable) {
                            cnt++;
                        }

                        // 开窗时间补充上
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        String stt = sdf.format(context.window().getStart());
                        String edt = sdf.format(context.window().getEnd());
                        // 转成实体对象
                        collector.collect(new BoundSort(stt, edt, key, cnt, null, null, null));
                    }
                });

        // 异步关联维度
        process.print();
        SingleOutputStreamOperator<BoundSort> asyncDS = AsyncDataStream.unorderedWait(process, new Mn6AsyncIOFunc(), 1000, TimeUnit.MILLISECONDS, 100);
        asyncDS.print();
        env.execute();

    }
}
